package org.dragonnova.iot.business.cloud.transport.service;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.haproxy.HAProxyMessageDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.internal.ThrowableUtil;

import java.io.IOException;

import org.dragonnova.iot.business.cloud.transport.adapter.HAProxyMessageDecoderAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

public class ProxyServer extends AbstractServer implements InitializingBean,
		DisposableBean {

	private final static Logger LOGGER = LoggerFactory
			.getLogger(NettyServer.class);

	private ChannelFuture serverChannel;
	private final EventExecutorGroup businessExecutorGroup;

	public ProxyServer(String host, int port) throws Exception {
		super(host, port);

		this.businessExecutorGroup = new DefaultEventExecutorGroup(10,
				new DefaultThreadFactory("businessExecutorGroup"));
		start();
	}

	@Override
	public void start() throws IOException {
		try {
			this.serverBootstrap.group(this.group, this.workGroup);
			if (this.osName.equals("Linux")) {
				this.serverBootstrap.channel(EpollServerSocketChannel.class);
			} else {
				this.serverBootstrap.channel(NioServerSocketChannel.class);
			}
			this.serverBootstrap.option(ChannelOption.SO_BACKLOG, 128)
					.option(ChannelOption.TCP_NODELAY, true)
					.option(EpollChannelOption.SO_REUSEPORT, true)
					.childOption(ChannelOption.SO_KEEPALIVE, true)
					.childHandler(new ChannelInitializer<SocketChannel>() {

						@Override
						protected void initChannel(SocketChannel arg0)
								throws Exception {
							arg0.pipeline().addLast(
									new IdleStateHandler(60, 120, 60));
							arg0.pipeline()
									.addLast(new HAProxyMessageDecoder());
							arg0.pipeline().addLast(businessExecutorGroup,
									"handler",
									new HAProxyMessageDecoderAdapter());
						}
					});

			this.serverChannel = this.serverBootstrap.bind(getBindAddress());
			this.serverChannel.addListener(new ChannelFutureListener() {
				@Override
				public void operationComplete(ChannelFuture future) {
					if (!future.isDone()) {
						LOGGER.error("Tag Server start Error. Socket Bind at \'"
								+ getBindAddress() + "\'}");
					} else {
						LOGGER.info("Tag Server started. Socket Bind at \'"
								+ getBindAddress() + "\'}");
					}
				}
			});

		} catch (Throwable e) {
			throw new IOException("start " + getClass().getName()
					+ " error when bind " + bindAddr + ". caused by "
					+ ThrowableUtil.stackTraceToString(e));
		}
	}

	@Override
	public void close() throws Exception {
		if (this.serverChannel != null) {
			this.serverChannel.channel().closeFuture().sync();
		}

		this.group.shutdownGracefully().sync();
		this.workGroup.shutdownGracefully().sync();

		this.businessExecutorGroup.shutdownGracefully();
	}

	@Override
	public void destroy() throws Exception {
		close();
	}

	@Override
	public void afterPropertiesSet() throws Exception {

	}
}